package ItemBased;

import java.io.IOException;
import java.util.Iterator;
import java.util.Vector;

import org.apache.hadoop.io.ArrayWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.Reducer;

public class ToItemVectors {

	
	public static class ToItemVectorsMap 
	extends MapReduceBase
    implements Mapper<LongWritable, Text, LongWritable, LongPairWritable>{
		
	      
	    public void map(LongWritable key, Text value, OutputCollector<LongWritable, LongPairWritable> output, Reporter reporter
	                    ) throws IOException {
	    	//System.out.println(key);
	    	String[] lines = value.toString().split(System.getProperty("line.separator"));
	    	for(String line : lines){
	    		String[] fields = line.split(",");
	    		LongWritable iID = new LongWritable();
	    		iID.set(Long.parseLong(fields[1]));
	    		LongPairWritable uIDRat = new LongPairWritable(Long.parseLong(fields[0]), Long.parseLong(fields[2]));	    		
	    		output.collect(iID, uIDRat);
	    	}
	    }		
	}

	public static class ToItemVectorsReduce
	extends MapReduceBase
    implements Reducer<LongWritable,LongPairWritable,LongWritable,LongPairArrayWritable> {
		
		public void reduce(LongWritable key, Iterator<LongPairWritable> values, 
				OutputCollector<LongWritable, LongPairArrayWritable> output, Reporter reporter
                ) throws IOException {
			Vector<LongPairWritable> list = new Vector<LongPairWritable> (); 
			while(values.hasNext()){
				list.add(values.next().get());
			}
			
			LongPairWritable[] pair = new LongPairWritable[list.size()];
			LongPairArrayWritable arrayPair = new LongPairArrayWritable();
			
			for(int i=0; i<list.size();i++){
				pair[i] = list.get(i);
			}
			arrayPair.set(pair);
			output.collect(key, arrayPair);
		}
	}
	
	
	/**
	 * @param args
	 */
	public static void main(String[] args) {
		// TODO Auto-generated method stub

	}

	
}
